package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func main() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	defer func(client sarama.SyncProducer) {
		err := client.Close()
		if err != nil {
			panic(err)
		}
	}(client)
	// 发送消息
	var message string
	fmt.Println("start send message:")
	for {
		_, err := fmt.Scanln(&message)
		if err != nil {
			fmt.Println(err)
			break
		}
		if message == "exit" {
			fmt.Println("bye!")
			break
		}
		// 构造一个消息
		msg := &sarama.ProducerMessage{}
		msg.Topic = "web_log"
		msg.Value = sarama.StringEncoder(message)
		pid, offset, err := client.SendMessage(msg)
		if err != nil {
			fmt.Println("send msg failed, err:", err)
			break
		}
		fmt.Printf("send message success, pid:%v offset:%v\n", pid, offset)
	}

}
